package com.css.resoft.p2pmonitor.dao.hbase;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.thrift2.generated.TColumnValue;
import org.apache.hadoop.hbase.thrift2.generated.TPut;
import org.apache.hadoop.hbase.thrift2.generated.TResult;
import org.apache.hadoop.hbase.util.Bytes;

import com.css.resoft.p2pmonitor.dao.TConnect;
import com.css.resoft.p2pmonitor.util.ReadConifgUtil;
import com.resoft.bigdata.p2p.record.MonthlyTodoRecord;

public class MonstatOp {
	
	public static String host = ReadConifgUtil.getInstance().getHost();
	public static int port = ReadConifgUtil.getInstance().getPort();
	
	public static boolean putData(MonthlyTodoRecord record) throws Exception{
		List<TColumnValue> columnValues = new ArrayList<TColumnValue>();
		TColumnValue columnValue = null;

		ByteBuffer table = ByteBuffer.wrap(record.getTableName().getBytes());
		byte []rowkey = record.getRowKey();
		String cl = record.getFamilies()[0];
		Map<byte[],byte[]> cell = record.getCells();
		if(cell == null) return false;
		
		for(Entry<byte[],byte[]> entry:cell.entrySet()){
			byte [] qualifier = entry.getKey();
			byte [] value = entry.getValue();
			columnValue = new TColumnValue();
			columnValue.setFamily(cl.getBytes());
			columnValue.setQualifier(qualifier);
			columnValue.setValue(value);
			columnValues.add(columnValue);
			
		}
		
		TPut put = new TPut();
		put.setRow(rowkey);
		put.setColumnValues(columnValues);
		
		TConnect con = new TConnect(host,port);
		con.putData(table, put);
		con.close();
		
		return true;
	}
	
	
	public List<MonthlyTodoRecord> getRecords() throws Exception{
		List<MonthlyTodoRecord> records = new ArrayList<MonthlyTodoRecord>();
		MonthlyTodoRecord record = null;
		
		TConnect con = new TConnect(host,port);
		
		long currentTime = System.currentTimeMillis();
		long startTs = currentTime - 30*24*3600*1000;
		String isDone = "NO";
		ByteBuffer table = ByteBuffer.wrap("MONSTAT_TODO".getBytes());
		
		List<TResult> results = con.getRecords(startTs, isDone, table);
		
		if(results != null){
			for (TResult tResult : results) {
				record = new MonthlyTodoRecord();
				record.setTs(Bytes.toLong(tResult.getRow()));
				record.setStoredRowKey(record.getRowKey());
				
				for(TColumnValue columnValue : tResult.getColumnValues()){
					byte [] qualifier = columnValue.getQualifier();
					byte [] value = columnValue.getValue();
					record.setValue(qualifier, value);
				}
				records.add(record);
			}
		}
		con.close();
		
		return records;
	}
	
	
	
}
